import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object WindowDemo {
  def main(args: Array[String]): Unit = {
     val conf = new SparkConf().setMaster("local[4]").setAppName("windowDemo")
     val context = new SparkContext(conf)
     context.setLogLevel("ERROR")
     //获取上下文对象
     val ssc = new StreamingContext(context , Seconds(5))

    val lines = ssc.socketTextStream("192.168.235.129",9997)

    val words = lines.flatMap( _.split(" "))

    val pairs = words.map( word=>(word,1))

   // val wordCount = pairs.reduceByKey( _ + _)

//    pairs.window(Seconds(10))
    val wordCount=pairs.reduceByKeyAndWindow((x:Int,y:Int)=>x+y,Seconds(10),Seconds(10))
    wordCount.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
